package function;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

public class HbaseSinkFunction extends RichSinkFunction<JSONObject> {
    Connection connection=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:192.168.1.10:2181");
    }

    @Override
    public void close() throws Exception {
        if (connection==null){
            connection.close();
        }
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        PreparedStatement preparedStatement = null;
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append("upsert into ");
        stringBuffer.append(value.getString("sinkdatabase")+".");
        stringBuffer.append(value.getString("sinktable")+"(");
        Set<String> after = value.getJSONObject("after").keySet();
        Collection<Object> vl = value.getJSONObject("after").values();
        stringBuffer.append(StringUtils.join(after,",")+") ");
        stringBuffer.append("values"+"('"+StringUtils.join(vl,"','")+"')");
        System.out.println("插入数据为》》》》》"+stringBuffer.toString());
        preparedStatement = connection.prepareStatement(stringBuffer.toString());
        preparedStatement.executeUpdate();
        connection.commit();
    }
}
